package com.atguigu.gmall.realtime.app.newapp

import com.atguigu.gmall.realtime.bean.CartInfo
import com.atguigu.gmall.realtime.handler.{CommonApp, DimHandler, EsHandler}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

import java.time.LocalDate

// cart_info 和 商品维表join  用户维表
object DwCartInfoApp extends CommonApp {

  def main(args: Array[String]): Unit = {
    val groupId: String = "dw_cart_info_app"
    val topic: String = "DWD_CART_INFO_I"
    val ssc: StreamingContext = initSsc(groupId, "local[4]", 5)

    val dStream: DStream[CartInfo] = getDstream[CartInfo](ssc, topic, groupId)

    val cartInfoDtDstream: DStream[CartInfo] = dStream.map { cartInfo =>
      val createTimeArr: Array[String] = cartInfo.create_time.split(" ")
      cartInfo.create_date = createTimeArr(0)
      val timeArr: Array[String] = createTimeArr(1).split(":")
      cartInfo.create_hour = timeArr(0)
      cartInfo
    }

    val dstreamWithU: DStream[CartInfo] = DimHandler.joinUser[CartInfo](cartInfoDtDstream,
      { cartInfo: CartInfo => cartInfo.user_id.toString }
    )

    val dstreamWithSku: DStream[CartInfo] = DimHandler.joinSku[CartInfo](dstreamWithU,
      { cartInfo: CartInfo => cartInfo.sku_id.toString }
    )
    dstreamWithSku.cache()
    dstreamWithSku.print(1000)


     EsHandler.saveDStreamToEs[CartInfo](dstreamWithSku,
          "gmall2022_cart_info_",
          "id",
          {Unit=>this.commitAllOffsetKafka()}
        )



    ssc.start()
    ssc.awaitTermination()

  }

}
